package cn.zwx.common.rocket.producer.listener;

import cn.zwx.common.rocket.producer.bo.TransactionListenerBO;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author zhangwenxue
 */
public class DefaultTransactionListener implements TransactionListener {

    private static final Logger logger = LoggerFactory.getLogger(DefaultTransactionListener.class);

    private Map<String, TransactionListener> listenerMap = new ConcurrentHashMap<>(8);

    public DefaultTransactionListener(List<TransactionListenerBO> transactionListenerList) {
        if (CollectionUtils.isEmpty(transactionListenerList)){
            logger.error("Transaction message  not setting TransactionListenerList , please check your config");
            return;
        }
        listenerMap = transactionListenerList.stream()
                .collect(Collectors.toConcurrentMap(TransactionListenerBO::getTopic,
                        TransactionListenerBO::getTransactionListener, (key1, key2) -> key2));

    }
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        final TransactionListener transactionListener = listenerMap.get(message.getTopic());
        if (Objects.isNull(transactionListener)){
            logger.error("local Transaction is not setting TransactionListener topic:{}",message.getTopic());
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return transactionListener.executeLocalTransaction(message,o);
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        final TransactionListener transactionListener = listenerMap.get(messageExt.getTopic());
        if (Objects.isNull(transactionListener)){
            logger.error("check Transaction is not setting TransactionListener topic:{}",messageExt.getTopic());
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return transactionListener.checkLocalTransaction(messageExt);
    }
}
